package com.com.zs21cp.watermark;


import com.com.zs21cp.ToolBean.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

/**
 * @program: flink
 * @description:
 * @author: Arctic
 * @create: 2024-03-12 21:28
 */
public class assignTimestamp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Event> source =
                (DataStreamSource<Event>) env.fromElements(
                        new Event("Marry","./home",100L),
                        new Event("alice","./home",100L),
                        new Event("Mark","./home",100L),
                        new Event("John","./home",100L)
                )
                        //有序流的watermark生成
//                        .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps()
//                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
//                            @Override
//                            public long extractTimestamp(Event element, long recordTimestamp) {
//                                return element.timestamp;
//                            }
//                        }));
                        //乱序流的waterMark生成,Duration.ofSecond(2)这是迟到时间为两秒的意思
        .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2))
        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
            @Override
            public long extractTimestamp(Event element, long recordTimestamp) {
                return element.timestamp;
            }
        }));


        env.execute();
    }
}